package com.lz.code.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @Author: liuzhi
 * @Date: 2022/3/9 22:19
 * @Description:
 */
@Component
public class MyConsumer {

    @KafkaListener(topics = "my-replicated-topic", groupId = "group1")
    public void listenGroup(ConsumerRecord<String,String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        // 手动提交offset
        ack.acknowledge();
    }

    @KafkaListener(topics = "my-replicated-topic", groupId = "group2")
    public void listenGroup2(ConsumerRecord<String,String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        // 手动提交offset
        ack.acknowledge();
    }

}
